package com.dpwgc.hdfs.service;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

/**
 * HDFS文件操作
 */
@Service
public class HdfsService {

    @Value("${hdfs.hdfsPath}")
    private String hdfsPath;

    @Value("${hdfs.hdfsName}")
    private String hdfsName;

    @Value("${hdfs.confPath}")
    private String confPath;

    //缓冲区大小
    private static final int bufferSize = 1024 * 1024 * 64;

    /**
     * 获取HDFS配置信息
     * @return
     */
    public Configuration getConfiguration() {

        Configuration conf = new Configuration();

        //加载配置
        conf.addResource(new Path(confPath + "core-site.xml"));
        conf.addResource(new Path(confPath + "hdfs-site.xml"));

        conf.set("dfs.client.use.datanode.hostname", "true");//设置按照hostName检索dataNode节点
        conf.set("dfs.replication","1");//设置优先级

        conf.set("fs.defaultFS", hdfsPath);
        return conf;
    }

    /**
     * 获取HDFS文件系统对象
     * @return FileSystem
     * @throws Exception
     */
    public FileSystem getFileSystem() throws Exception {
        //在构造FileSystem对象时，将hdfsName用户名参数传递进去
        return FileSystem.get(new URI(hdfsPath), getConfiguration(), hdfsName);
    }

    /**
     * 创建文件夹
     * @param path 文件夹路径
     * @return
     * @throws Exception
     */
    public boolean mkdir(String path) throws Exception {
        if (StringUtils.isEmpty(path)) {
            return false;
        }
        if (existFile(path)) {
            return true;
        }
        FileSystem fs = getFileSystem();
        // 目标路径
        Path srcPath = new Path(path);
        boolean flag = fs.mkdirs(srcPath);
        fs.close();
        return flag;
    }

    /**
     * 读取当前目录信息
     * @param path 文件夹路径
     * @return
     * @throws Exception
     */
    public List<Map<String, Object>> readPathInfo(String path) throws Exception {
        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }
        FileSystem fs = getFileSystem();
        // 目标路径
        Path newPath = new Path(path);
        FileStatus[] statusList = fs.listStatus(newPath);
        List<Map<String, Object>> list = new ArrayList<>();
        if (null != statusList && statusList.length > 0) {
            for (FileStatus fileStatus : statusList) {
                Map<String, Object> map = new HashMap<>();
                map.put("filePath", fileStatus.getPath());
                map.put("fileStatus", fileStatus.toString());
                list.add(map);
            }
            return list;
        } else {
            return null;
        }
    }

    /**
     * 获取文件在集群中的位置
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public BlockLocation[] getFileBlockLocations(String path) throws Exception {
        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }
        FileSystem fs = getFileSystem();
        //目标路径
        Path srcPath = new Path(path);
        FileStatus fileStatus = fs.getFileStatus(srcPath);
        return fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
    }

    /**
     * 文件上传
     * @param path 文件保存路径
     * @param file 文件
     * @throws Exception
     */
    public boolean uploadFile(String path, MultipartFile file) throws Exception {

        String fileName = file.getOriginalFilename();

        FileSystem fs = getFileSystem();

        try{

            //拼接文件路径
            Path newPath = new Path(path + "/" + fileName);
            System.out.println(newPath);
            //打开输出流
            FSDataOutputStream outputStream = fs.create(newPath);
            outputStream.write(file.getBytes());
            outputStream.close();
            fs.close();
            return true;

        }catch (Exception e){

            fs.close();
            return false;

        } finally {
            fs.close();
        }
    }

    /**
     * 读取文件内容
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public String readFile(String path) throws Exception {
        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }
        FileSystem fs = getFileSystem();
        //目标路径
        Path srcPath = new Path(path);
        FSDataInputStream inputStream = null;
        try {
            inputStream = fs.open(srcPath);
            //防止中文乱码
            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
            String lineTxt = "";
            StringBuffer stringBuffer = new StringBuffer();
            while ((lineTxt = reader.readLine()) != null) {
                stringBuffer.append(lineTxt);
            }
            return stringBuffer.toString();
        } finally {
            inputStream.close();
            fs.close();
        }
    }

    /**
     * 文件下载（返回字符流数据给前端）
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public char[] downloadFile(String path) throws Exception {
        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }
        FileSystem fs = getFileSystem();
        //目标路径
        Path srcPath = new Path(path);

        try {
            FSDataInputStream inputStream = fs.open(srcPath);
            byte[] bytes = IOUtils.readFullyToByteArray(inputStream);

            //字节流转字符流
            Charset cs = Charset.forName("UTF-8");
            ByteBuffer bb = ByteBuffer.allocate(bytes.length);
            bb.put(bytes).flip();
            CharBuffer cb = cs.decode(bb);
            return cb.array();

        } finally {
            fs.close();
        }
    }

    /**
     * 读取文件夹下的文件列表
     * @param path 文件夹路径
     * @return
     * @throws Exception
     */
    public List<Map<String, String>> listFile(String path) throws Exception {
        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }

        FileSystem fs = getFileSystem();
        //目标路径
        Path srcPath = new Path(path);
        //递归找到所有文件
        RemoteIterator<LocatedFileStatus> filesList = fs.listFiles(srcPath, true);

        List<Map<String, String>> returnList = new ArrayList<>();

        while (filesList.hasNext()) {
            LocatedFileStatus next = filesList.next();
            String fileName = next.getPath().getName();
            Path filePath = next.getPath();
            Map<String, String> map = new HashMap<>();
            map.put("fileName", fileName);
            map.put("filePath", filePath.toString());
            returnList.add(map);
        }

        fs.close();
        return returnList;
    }

    /**
     * 重命名文件
     * @param oldPath 旧文件路径
     * @param newPath 新文件路径
     * @return
     * @throws Exception
     */
    public boolean renameFile(String oldPath, String newPath) throws Exception {
        if (StringUtils.isEmpty(oldPath) || StringUtils.isEmpty(newPath)) {
            return false;
        }
        FileSystem fs = getFileSystem();

        boolean flag = fs.rename(new Path(oldPath), new Path(newPath));
        fs.close();
        return flag;
    }

    /**
     * 删除文件
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public boolean deleteFile(String path) throws Exception {
        if (StringUtils.isEmpty(path)) {
            return false;
        }
        if (!existFile(path)) {
            return false;
        }
        FileSystem fs = getFileSystem();
        Path srcPath = new Path(path);
        boolean flag = fs.deleteOnExit(srcPath);
        fs.close();
        return flag;
    }

    /**
     * 复制文件
     * @param sourcePath 原文件路径
     * @param targetPath 目标文件路径
     * @throws Exception
     */
    public void copyFile(String sourcePath, String targetPath) throws Exception {
        if (StringUtils.isEmpty(sourcePath) || StringUtils.isEmpty(targetPath)) {
            return;
        }
        FileSystem fs = getFileSystem();
        //原文件路径
        Path oldPath = new Path(sourcePath);
        //目标文件路径
        Path newPath = new Path(targetPath);

        FSDataInputStream inputStream = null;
        FSDataOutputStream outputStream = null;
        try {
            inputStream = fs.open(oldPath);
            outputStream = fs.create(newPath);

            IOUtils.copyBytes(inputStream, outputStream, bufferSize, false);
        } finally {
            inputStream.close();
            outputStream.close();
            fs.close();
        }
    }

    /**
     * 查看文件是否已存在
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public boolean existFile(String path) throws Exception {
        if (StringUtils.isEmpty(path)) {
            return false;
        }
        FileSystem fs = getFileSystem();

        Path srcPath = new Path(path);
        return fs.exists(srcPath);
    }
}
